package com.dianpingmedia.common;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 云平台通用类----发送队列(需要发送的队列和收到结果需要返回的队列)
 *
 * @author Karl He
 * @version 1.0
 * @Date 2018-10-26
 */
public class BaseSendQueue<T> {
    /**
     * 进入队列元素的数量
     */
    private long counterIn = 0L;
    /**
     * 取走队列元素的数量(包括超时)
     */
    private long counterOut = 0L;
    private BlockingQueue<T> queue = new LinkedBlockingQueue<T>();

    public BaseSendQueue() {

    }

    public long getSize() {
        return (counterIn - counterOut);
    }

    public long getCouterIn() {
        return counterIn;
    }

    public long getCouterOut() {
        return counterOut;
    }

    /**
     * 强制清空队列并重新计数
     */
    public void reset() {
        queue.clear();
        counterOut = 0L;
        counterIn = 0L;
    }

    /**
     * 安全的重置队列计数
     */
    public void resetIfEmpty() {
        Thread t = new Thread(new Runnable() {
            @Override
            public void run() {
                for (; ; ) {
                    if (getSize() == 0) {
                        counterOut = 0L;
                        counterIn = 0L;
                    }
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (Exception e) {
                    }
                }

            }
        });
        t.start();
    }

    /**
     * 单个元素入队列
     *
     * @return
     */
    public boolean enqueue(T t) {
        if (queue.offer(t)) {
            counterIn++;
            return true;
        }
        return false;
    }

    /**
     * 批量入队列
     * @return
     */
    public boolean enqueue(List<T> ts) {
        int count = 0;
        for (T t : ts) {
            if (queue.offer(t)) {
                count++;
            }
        }
        counterIn += count;
        return (count > 0);
    }

    /**
     * 从队列获取单个数据,即时返回
     *
     * @return
     */
    public T dequeue() {
        T t = queue.poll();
        if (t != null) {
            counterOut++;
        }

        return t;
    }

    /**
     * 从队列获取批量数据,即时返回
     *
     * @param maxCount
     * @return
     */
    public List<T> dequeue(int maxCount) {
        T t = queue.poll();
        if (t == null) {
            return null;
        }

        List<T> list = new ArrayList<>(1);
        list.add(t);

        try {
            queue.drainTo(list, maxCount - 1);
        } catch (Exception e) {
        }

        int count = list.size();

        counterOut += count;

        return list;
    }


    /**
     * 从队列获取批量数据,堵塞
     *
     * @param maxCount      返回的最大条数
     * @param maxWaitMillis 从队列连续读取,最大等待的毫秒数
     * @return
     */
    public List<T> dequeueWait(int maxCount, int maxWaitMillis) {
        List<T> list = null;
        try {
            T t = queue.take();

            if (t == null) {
                return null;
            }

            list = new ArrayList<T>(1);
            list.add(t);
            int count = 1;

            long start = System.currentTimeMillis();

            try {
                for (; ; ) {
                    t = queue.poll(maxWaitMillis, TimeUnit.MILLISECONDS);
                    if (t == null) {
                        break;
                    }

                    list.add(t);
                    if (++count >= maxCount) {
                        break;
                    }

                    if (System.currentTimeMillis() - start > maxWaitMillis) {
                        break;
                    }
                }
            } catch (Exception e) {
            }

            counterOut += count;

        } catch (Exception e) {

        }
        return list;
    }
}
